package com.niit.mapFile.write;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;

import java.io.IOException;
import java.util.Iterator;
import java.util.Random;

/**
 * Date:2024/10/18
 * Author：Ys
 * Description:
 */
public class WriteMapFile {

    //Driver
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //用于管理当前程序的配置对象
        Configuration conf = new Configuration();
        //实例化 作业
        Job job = Job.getInstance(conf, WriteMapFile.class.getName());
        //设置作业的jar包
        job.setJarByClass(WriteMapFile.class);
        //设置作业的输入格式 TextInputFormat
        job.setInputFormatClass(TextInputFormat.class);
        //设置作业的输入路径
        FileInputFormat.addInputPath(job,new Path("input/file"));

        //设置Mapper端的实现类 以及输出的 Key Value
        job.setMapperClass(WriteMapper.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);

        //设置Reducer端的实现类 以及输出的 Key Value
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        //输出格式 MapFileOutputFormat
        job.setOutputFormatClass(MapFileOutputFormat.class);

        Path output = new Path("map_write");
        output.getFileSystem(conf).delete(output,true);

        FileOutputFormat.setOutputPath(job,output);

        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }

    /*
     定义Mapper类

        MapFile特殊处理后的文件 有两个文件  一个索引文件  一个数据文件
                                                   KEYIN      VALUEIN   KEYOUT    VALUEOUT
     */
    public static class WriteMapper extends Mapper<LongWritable, Text, IntWritable,Text>{

        //定义要输出的Key，每次随机生成一个值
        private IntWritable keyOut = new IntWritable();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {
            //随机生成一个值
            Random random = new Random();
            keyOut.set(random.nextInt(99999));
                        //value  是个什么东西？  每一行数据
            //context: 上下文： 承上（数据文件）   启下（Reducer）    的作用    初中语文
            context.write(keyOut,value);
        }
    }


    /*
      定义一个Reducer类

                                                    KETIN  VALUEIN         KEYOUT VALUEOUT
     */
    public static class WriteReducer extends Reducer<IntWritable,Text,IntWritable,Text>{

        //                                        values 一组数据 List 数组
        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Reducer<IntWritable, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {

            Iterator<Text> iterator = values.iterator();//迭代器
            //迭代器看作是要一个有规则顺序准满砖的水桶，只能自上而下的去拿板砖
            while (iterator.hasNext()){// 当前迭代器是否有新（下一个）数据   是：true  否:false
                Text value = iterator.next(); //移动指针得到迭代器中的一个数据
                //context: 承上（Mapper）          启下（文件）
                context.write(key,value);
            }


        }
    }


}
